package com.wudl.flink.sql;

import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @ClassName : Flink_SQL_join
 * @Description : Flink -sql  动态表关联
 * @Author :wudl
 * @Date: 2021-08-20 00:16
 */

public class Flink_SQL_join {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //默认值为0   表示FlinkSQL中的状态永久保存
        System.out.println(tableEnv.getConfig().getIdleStateRetention());

        //执行FLinkSQL状态保留10秒
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(10));


        //2.读取端口数据创建流
        SingleOutputStreamOperator<TableA> aDS = env.socketTextStream("192.168.1.180", 8888)
                .map(line -> {
                    String[] split = line.split(",");
                    return new TableA(split[0], split[1]);
                });
        SingleOutputStreamOperator<TableB> bDS = env.socketTextStream("192.168.1.180", 9999)
                .map(line -> {
                    String[] split = line.split(",");
                    return new TableB(split[0], Integer.parseInt(split[1]));
                });

        // 3. 将流转化为动态表
        tableEnv.createTemporaryView("tabA", aDS);
        tableEnv.createTemporaryView("tabB", bDS);
        // 4. 双流join
        tableEnv.sqlQuery("select * from tabA a left join tabB b on a.id = b.id").execute().print();

        env.execute();

    }
}
